package cn.mjnxin.iot.gate.worker;

import cn.mjnxin.iot.base.domain.ChannelData;
import cn.mjnxin.iot.base.log.LogWorker;
import cn.mjnxin.iot.base.process.DataProcessWorker;
import cn.mjnxin.iot.base.util.MixUtil;
import cn.mjnxin.iot.gate.cache.GateRuntimeCache;
import cn.mjnxin.iot.gate.cache.GateRuntimeCache.HostChannel;
import io.netty.channel.Channel;

/**
 * 上行报文中转
 * 
 * @Description: 终端经过网关中转实际发送前置
 * @author CLG 2023年10月10日
 * @version V1.0.0
 */
public class Gate2HostWorker extends DataProcessWorker<ChannelData> {
	private static Gate2HostWorker worker    = null;
	
	//非单例模式：调用一次就初始化一次
	public static void init(int poolSize) throws Exception {
		worker = new Gate2HostWorker(poolSize);
	}

	public static Gate2HostWorker getInstance() {
		return worker;
	}

	private Gate2HostWorker(int poolSize) throws Exception {
		super(poolSize, "Gate2HostWorker");
		this.start();
	}

	@Override
	protected void proccess(ChannelData channelData) throws Exception {
		// 获取前置与网关连接channel
		HostChannel hostChannel = GateRuntimeCache.choiceMngHostChannel(channelData.getTerminalAddress(), channelData.getPid());
		if (hostChannel != null) {
			channelData.setHostAddress(hostChannel.getHostKey());
			Channel channel = hostChannel.getChannel();
			if (channel.isActive() && channel.isWritable()) {
				channel.writeAndFlush(channelData);
				return;
			}

			LogWorker.logForce(String.format("%d|%s MngHost[%s]不可写", channelData.getPid(), channelData.getTerminalAddress(), hostChannel.getHostKey()));
		}

		hostChannel = GateRuntimeCache.choiceHostChannel(channelData.getPid(), channelData.getTerminalAddress());
		/**
		 * 1.通过channel发送数据--会执行编码器等handler
		 * 2.发送数据前判断masterChannel是否可写，因为配置了高水位和低水位，有可能channel为“不可写”状态
		 */
		if (hostChannel != null) {
			channelData.setHostAddress(hostChannel.getHostKey());
			Channel channel = hostChannel.getChannel();
			if (channel.isActive() && channel.isWritable()) {
				channel.writeAndFlush(channelData);
				return;
			}

			LogWorker.logForce(String.format("%d|%s Host[%s]不可写", channelData.getPid(), channelData.getTerminalAddress(), hostChannel.getHostKey()));
			byte[] dst = new byte[channelData.getSocketData().getByteBuf().readableBytes()];
			channelData.getSocketData().getByteBuf().readBytes(dst);
			LogWorker.logForce(String.format("NS T[%s]->G->H[%s] = %s", channelData.getTerminalAddress(), hostChannel.getHostKey(), MixUtil.bytes2hex(dst)));
			channelData.getSocketData().getByteBuf().readerIndex(0);
		} else {
			HostChannel mngChannel = GateRuntimeCache.getMngChannel(channelData.getPid());
			if (mngChannel != null) {
				Channel channel = mngChannel.getChannel();
				channelData.setHostAddress(mngChannel.getHostKey());
				if (channel.isActive() && channel.isWritable()) {
					LogWorker.logForce(String.format("使用备用MngHost[%s]", mngChannel.getHostKey()));
					channel.writeAndFlush(channelData);
					return;
				}

				LogWorker.logForce(String.format("%d|%s 备用MngHost[%s]不可写", channelData.getPid(), channelData.getTerminalAddress(), mngChannel.getHostKey()));
			}

			byte[] dst = new byte[channelData.getSocketData().getByteBuf().readableBytes()];
			channelData.getSocketData().getByteBuf().readBytes(dst);
			if (mngChannel != null) {
				LogWorker.logForce(String.format("NS T[%s]->G->H[%s] = %s", channelData.getTerminalAddress(), mngChannel.getHostKey(), MixUtil.bytes2hex(dst)));
			} else {
				LogWorker.logForce(String.format("NS T[%s]->G->H = %s", channelData.getTerminalAddress(), MixUtil.bytes2hex(dst)));
			}
			channelData.getSocketData().getByteBuf().readerIndex(0);
		}
	}
}
